package com.atguigu.gmall.realtime.dws.app;

import com.atguigu.gmall.realtime.common.base.BaseSQLApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import com.atguigu.gmall.realtime.dws.function.KeywordUDTF;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Felix
 * @date 2024/6/26
 * 搜索关键词聚合统计
 * 需要启动的进程
 *      zk、kafka、flume、doris、DwdBaseLog、DwsTrafficSourceKeywordPageViewWindow
 */
public class DwsTrafficSourceKeywordPageViewWindow extends BaseSQLApp {
    public static void main(String[] args) {
        new DwsTrafficSourceKeywordPageViewWindow().start(
                10021,
                4,
                "dws_traffic_source_keyword_page_view_window"
        );

    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        //TODO 注册分词函数到表执行环境中
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);
        //TODO 从页面日志主题中读取数据  创建动态表 指定Watermark的生成策略并提取事件时间字段
        tableEnv.executeSql("CREATE TABLE page_log (\n" +
                "    common map<string,string>,\n" +
                "    page map<string,string>,\n" +
                "    ts bigint ,\n" +
                "    et as TO_TIMESTAMP_LTZ(ts, 3),\n" +
                "    WATERMARK FOR et AS et\n" +
                ")" + SQLUtil.getKafkaDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE,"dws_traffic_source_keyword_page_view_window"));
        //tableEnv.executeSql("select * from page_log").print();

        //TODO 过滤出搜索行为
        Table searchTable = tableEnv.sqlQuery("select\n" +
                "   page['item'] fullword,\n" +
                "   et " +
                " from page_log where page['last_page_id']='search' and page['item_type']='keyword' " +
                " and page['item'] is not null");
        //searchTable.execute().print();
        tableEnv.createTemporaryView("search_table",searchTable);

        //TODO 分词 并和原表字段进行关联
        Table splitTable = tableEnv.sqlQuery("SELECT\n" +
                "    keyword,et\n" +
                "FROM search_table,LATERAL TABLE(ik_analyze(fullword)) t(keyword)");
        //splitTable.execute().print();
        tableEnv.createTemporaryView("split_table",splitTable);

        //TODO 分组、开窗、聚合计算
        Table resTable = tableEnv.sqlQuery("SELECT\n" +
                "    DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') stt,\n" +
                "    DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') edt,\n" +
                "    DATE_FORMAT(window_start, 'yyyy-MM-dd') cur_date, \n" +
                "    keyword,\n" +
                "    count(*) keyword_count\n" +
                "  FROM TABLE(\n" +
                "    TUMBLE(TABLE split_table, DESCRIPTOR(et), INTERVAL '10' second))\n" +
                "  GROUP BY keyword,window_start, window_end");

        //resTable.execute().print();

        //TODO 将聚合的结果写到Doris
        tableEnv.executeSql("create table dws_traffic_source_keyword_page_view_window(\n" +
                "     stt string,\n" +
                "     edt string,\n" +
                "     cur_date string,\n" +
                "     keyword string,\n" +
                "     keyword_count bigint\t\n" +
                ")" + SQLUtil.getDorisDDL("dws_traffic_source_keyword_page_view_window"));
        resTable.executeInsert("dws_traffic_source_keyword_page_view_window");
    }
}
